package com.poetic.order;

import com.poetic.order.domain.OrderEvent;
import com.poetic.order.domain.OrderResult;
import com.poetic.order.function.OrderPayMatch;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * <pre>
 *  我们同样可以利用 Process Function，自定义实现检测订单超时的功能。为了简
 * 化问题，我们只考虑超时报警的情形，在 pay 事件超时未发生的情况下，输出超时
 * 报警信息。
 * 一个简单的思路是，可以在订单的 create 事件到来后注册定时器，15 分钟后触
 * 发；然后再用一个布尔类型的 Value 状态来作为标识位，表明 pay 事件是否发生过。
 * 如果 pay 事件已经发生，状态被置为 true，那么就不再需要做什么操作；而如果 pay
 * 事件一直没来，状态一直为 false，到定时器触发时，就应该输出超时报警信息。
 * Created by lianghuikun on 2020-09-16.
 * </pre>
 *
 * @author lianghuikun
 */
public class OrderTimeoutWithoutCepTask {
    public static void main(String[] args) throws Exception {
        String path = "/Users/lianghuikun/indigo/poetic/flink/user-behavior-anylysis/data/OrderLog.csv";

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);
        // 读取订单事件流
        KeyedStream<OrderEvent, String> orderEventStream = env.readTextFile(path)
                .map(new MapFunction<String, OrderEvent>() {
                    @Override
                    public OrderEvent map(String data) throws Exception {
                        String[] line = data.split("\\,");

                        return new OrderEvent(Long.valueOf(line[0]),
                                line[1],
                                line[2],
                                Long.valueOf(line[3]));
                    }
                })
                .filter(new FilterFunction<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent value) throws Exception {
                        return StringUtils.isNotBlank(value.getTxId());
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
                            @Override
                            public long extractTimestamp(OrderEvent element, long recordTimestamp) {
                                return element.getEventTime() * 1000;
                            }
                        })).keyBy(new KeySelector<OrderEvent, String>() {
                    @Override
                    public String getKey(OrderEvent value) throws Exception {
                        return value.getTxId();
                    }
                });
        OutputTag<OrderResult> orderTimeoutOutputTag = new OutputTag<OrderResult>("orderTimeout"){};

//            orderEventStream.print();
        SingleOutputStreamOperator<OrderResult> orderResultStream = orderEventStream.process(new OrderPayMatch(orderTimeoutOutputTag));


        orderResultStream.print("payed");
        orderResultStream.getSideOutput(orderTimeoutOutputTag).print("timeout");
        env.execute("order timeout without cep job");

    }
}
